package com.dy.realtime.dwd.app;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd.app
 * @className: DwdTradeConsultationPaySuc
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/10 20:52
 * @version: 1.0
 * todo 交易域问诊支付成功事务事实表
 */

public class DwdTradeConsultationPaySuc extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeConsultationPaySuc().start(Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC, Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC,4,10012);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv,Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC);
        Table table = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['update_time'] consultation_pay_suc_time,\n" +
                "    `data`['consultation_fee'] consultation_fee,\n" +
                "    `data`['status'] status,\n" +
                "    `data`['doctor_id'] doctor_id,\n" +
                "    `data`['patient_id'] patient_id,\n" +
                "    `data`['user_id'] user_id\n" +
                "from topic_db_medical\n" +
                "where `table` = 'consultation'\n" +
                "  and `type` = 'update'\n" +
                "  and `data`['status'] = '203'");
//        table.execute().print();
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC+"\n" +
                "(\n" +
                "    id                        string,\n" +
                "    create_time               string,\n" +
                "    consultation_pay_suc_time string,\n" +
                "    consultation_fee          string,\n" +
                "    status                    string,\n" +
                "    doctor_id                  string,\n" +
                "    patient_id                string,\n" +
                "    user_id                   string\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC));
        table.executeInsert(Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC);
    }
}
